package com.xr.redis.pubsub.service.impl;

import com.xr.redis.pubsub.model.Message;
import com.xr.redis.pubsub.service.IPubSubServiceAdapter;
import com.xr.redis.pubsub.service.listener.AbstractPubSubListener;
import com.xr.redis.util.JSONUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Arnold Yand
 * @summary 基于redis发布订阅实现的适配器
 * @time 2018/8/9 18:04
 */
public class PubSubImplementAdapter implements IPubSubServiceAdapter {

  protected Logger logger = LoggerFactory.getLogger(PubSubImplementAdapter.class);

  // redis订阅监听器
  private JedisPubSub pubSubListener;

  private ConcurrentHashMap<String, List<AbstractPubSubListener>> channelMapToListener;

  private Jedis jedisClient;

  public PubSubImplementAdapter(Jedis jedisClient, ConcurrentHashMap<String, List<AbstractPubSubListener>> channelMapToListener){
    this.jedisClient = jedisClient;
    this.channelMapToListener = channelMapToListener;
  }

  @Override
  public void init() throws Exception {
    // 创建监听器
    pubSubListener = new JedisPubSub(){
      @Override
      public void onMessage(String channel, String message) {
        try{
          List<AbstractPubSubListener> listeners = channelMapToListener.getOrDefault(channel, Collections.EMPTY_LIST);
          for(AbstractPubSubListener l : listeners){
            l.onMessage(JSONUtil.parseObject(message, Message.class));
          }
        }catch (Exception e){
          logger.error("process message failed, channel:{}, message:{}", channel, message);
        }
      }
      @Override
      public void onSubscribe(String channel, int subscribedChannels) {
        logger.info("start subscribe channel {}, and total subscribed {} channels.", channel, subscribedChannels);
      }

      @Override
      public void onUnsubscribe(String channel, int subscribedChannels) {
        logger.info("un-subscribe channel {}, and total subscribed {} channels.", channel, subscribedChannels);
      }
    };

    // 订阅频道
    for(String chn : channelMapToListener.keySet()){
      subscribeChannel(chn);
    }
  }

  @Override
  public void publish(String channel, Message message) throws Exception {
    if(channel != null && message != null){
      message.setChannel(channel);
      this.jedisClient.publish(channel, JSONUtil.toJSONString(message));
    }
  }

  @Override
  public void subscribe(String channel) throws Exception {
    subscribeChannel(channel);
  }

  private void subscribeChannel(String channel){
    if(pubSubListener != null && jedisClient != null){
      jedisClient.subscribe(pubSubListener, channel);
    }
  }

}
